package com.cn.file;

import com.cn.common.BrokerEntity;
import com.cn.common.ServerThread;
import com.cn.file.index.IndexService;
import com.cn.store.Controller;

import java.io.File;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 * 文件存储
 */
public class DefaultMessageStore {

    //commitlog 消息真实的存储文件
    private final CommitLog commitLog;

    //consume 消息在commitlog中的位置
    private final ConcurrentMap<String /*topic*/, ConcurrentMap<Integer /*queueid*/, ConsumeQueue /*consume*/>> consumeQueueTables;

    //consume存储服务
    private ReputMessageService reputMessageService = null;
    private FlushQueueService flushQueueService;
    private final MessageStoreConfig messageStoreConfig;
    private boolean start = false;

    private final List<DispatcherService> dispatcherServices = new ArrayList<>();
    private IndexService indexService;

    private final ConcurrentMap<String /*cluster*/, Set<BrokerEntity>> clusterTables = new ConcurrentHashMap<>();
    private final Controller brokerController;

    public DefaultMessageStore(MessageStoreConfig messageStoreConfig, Controller brokerController) {
        this.brokerController = brokerController;
        this.messageStoreConfig = messageStoreConfig;
        this.commitLog = new CommitLog(this);
        this.consumeQueueTables = new ConcurrentHashMap<>(128);
        this.reputMessageService = new ReputMessageService();
        this.flushQueueService = new FlushQueueService();

        indexService = new IndexService(this.messageStoreConfig.getIndexPath());
        dispatcherServices.add(indexService);
    }

    public void load() {
        //加载并恢复consumequeue文件
        this.loadConsumeQueue();

        //恢复index数据文件
        this.recoverIndex();

        //加载commitlog数据文件
        this.commitLog.load();

        //恢复数据文件缓存位置、可写入位置
        this.recover();

        start = true;
    }

    /**
     * 加载consumequeue文件
     */
    public void loadConsumeQueue() {
        /**
         * 加载consumequeue文件，并放到缓存中
         */

        // 1. 遍历consumequeue文件夹
        File dirLogic = new File(this.messageStoreConfig.getStoreConsumePath());
        if (dirLogic != null && dirLogic.exists()) {
            //遍历topic文件夹
            for (File file : dirLogic.listFiles()) {
                String topic = file.getName();
                //遍历队列
                for (File f : file.listFiles()) {
                    //队列id
                    int queueId = Integer.valueOf(f.getName()).intValue();
                    ConsumeQueue consumeQueue = new ConsumeQueue(this, this.messageStoreConfig.getConsumeFileSize(), topic, queueId, this.messageStoreConfig.getStoreConsumePath());
                    this.putConsumeQueue(topic, queueId, consumeQueue);
                    consumeQueue.load();
                }
            }
        }
    }

    protected void putConsumeQueue(String topic, int queueId, ConsumeQueue consumeQueue) {

        //判断是否已经存在
        ConcurrentMap<Integer, ConsumeQueue> queue = this.consumeQueueTables.get(topic);
        if (null == queue) {
            queue = new ConcurrentHashMap<>(128);
            queue.put(queueId, consumeQueue);
            this.consumeQueueTables.put(topic, queue);
        } else {
            queue.put(queueId, consumeQueue);
        }
    }

    public void start() {
        this.commitLog.start();
        this.reputMessageService.start();
        this.flushQueueService.start();
    }

    public void stop() {
        this.commitLog.stop();
        this.reputMessageService.stop();
    }

    public void recover() {
//        reputMessageService.setReputFromOffset(this.commitLog.getMaxOffset());

        //恢复consumequeue队列数据文件
        this.recoverConsume();

        //恢复commitlog数据文件
        this.recoverNormal();

        //恢复topic
        this.recoverTopicQueueTable();
    }

    public void recoverConsume() {
        for (ConcurrentMap<Integer, ConsumeQueue> map : this.consumeQueueTables.values()) {
            for (ConsumeQueue consumeQueue : map.values()) {
                consumeQueue.recover();
            }
        }
    }

    public void recoverNormal() {
        this.commitLog.recover();
    }

    public void recoverIndex() {
        indexService.load();
    }

    /**
     * 逻辑队列当前已经写入了多少个消息了
     */
    public void recoverTopicQueueTable() {
        Map<String, Long> map = new HashMap<>(1024);
        for (ConcurrentMap<Integer, ConsumeQueue> maps : this.consumeQueueTables.values()) {
            for (ConsumeQueue consumeQueue : maps.values()) {
                long offset = consumeQueue.getConsumeQueueOffset();
                map.put(consumeQueue.getTopic() + "-" + consumeQueue.getQueueId(), offset);
            }
        }
        this.commitLog.setTopicQueueTable(map);
    }

    public boolean putMessage(long offset, byte[] data) {
        if (!start || data == null || data.length == 0) return false;
        return this.commitLog.putMessage(offset, data);
    }

    public AppendMessageResult putMessage(MessageStore messageStore) {
        if (!start) {
            return new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR, 0, 0, null, 0, 0, 0, 0);
        }
        return this.commitLog.putMessage(messageStore);
    }


    /**
     * consumequeue队列消息存储
     *
     * @param dispatchRequest
     */
    public void doDispatch(DispatchRequest dispatchRequest) {
        try {
            //保存逻辑队列消息
            ConsumeQueue consumeQueue = this.findConsumeQueue(dispatchRequest.getTopic(), dispatchRequest.getQueueId());
//            System.out.println("第：" + consumeQueue.getConsumeQueueOffset() + "条消息，队列：" + consumeQueue.getQueueId() + "===" + dispatchRequest.getCommitLogOffset());
            consumeQueue.putMessage(dispatchRequest);
        } catch (Exception e) {
//            e.printStackTrace();
        }

        try {
            indexService.dispatch(dispatchRequest);
        } catch (Exception e) {
//            e.printStackTrace();
        }
    }

    /**
     * 查找队列是否存在
     *
     * @param topic
     * @param queueId
     * @return
     */
    public ConsumeQueue findConsumeQueue(String topic, int queueId) {
        ConcurrentMap<Integer, ConsumeQueue> map = this.consumeQueueTables.get(topic);
        if (null == map) {
            //该主题的队列不存在

            //新队列
            ConcurrentMap<Integer, ConsumeQueue> tmpMap = new ConcurrentHashMap<>(128);
            //旧队列
            ConcurrentMap<Integer, ConsumeQueue> oldMap = new ConcurrentHashMap<>();
            //如果存在，那么返回旧队列，如果不存在，那么保存新队列
            oldMap = this.consumeQueueTables.putIfAbsent(topic, tmpMap);
            if (null != oldMap) {
                map = oldMap;
            } else {
                map = tmpMap;
            }
        }

        //逻辑队列
        ConsumeQueue logic = map.get(queueId);
        if (null == logic) {
            //新建
            logic = new ConsumeQueue(this, this.messageStoreConfig.getConsumeFileSize(), topic, queueId, this.messageStoreConfig.getStoreConsumePath());
            map.put(queueId, logic);
        }
        return logic;
    }

    /**
     * @param topic
     * @param queueId
     * @param offset  -- 在queue队列中的第几个消息（offset * 20 才表示真实的位移，offset=0表示第一个消息, ......）
     * @return
     */
    public GetMessageResult getMessage(String topic, int queueId, long offset) {
        long nextOffset = offset;
        //查询队列
        ConsumeQueue consumeQueue = findConsumeQueue(topic, queueId);
        if (consumeQueue != null) {
            long minOffset = 0l;
            long maxOffset = 0l;

            GetMessageResult messageResult = new GetMessageResult();
            messageResult.setMessageStatus(GetMessageStatus.NOT_FOUND);

            SelectMappedBufferResult selectMappedBufferResult = consumeQueue.getIndexBuffer((int) offset);
            if (selectMappedBufferResult == null) {
                return new GetMessageResult(GetMessageStatus.NOT_FOUND, 0, 0);
            }

            int size = 0;
            int i = 0;
            for (; i < selectMappedBufferResult.getSize(); i += ConsumeQueue.UNIT_SIZE) {
                //每次最多拉取50条，剩下的在下一次拉取
                if (size >= 1000) {
                    break;
                }
                ByteBuffer byteBuffer = selectMappedBufferResult.getByteBuffer();
                if (byteBuffer == null) {
                    break;
                }
                //消息在commitlog文件中的真实位移
                long offset1 = byteBuffer.getLong();
                //消息大小
                int msgSize1 = byteBuffer.getInt();
                //临时字段（暂时不用）
                long tagsCode1 = byteBuffer.getLong();

                //查询消息在commitlog中的真实位置
                SelectMappedBufferResult result = this.commitLog.getMessage(offset1, msgSize1);
                if (result != null) {
//                    System.out.println("新的消息：" + offset1);
                    messageResult.addMessage(result);
                    messageResult.setMessageStatus(GetMessageStatus.FOUND);
                } else {
//                    System.out.println("没有！");
                }
                size++;
            }

            nextOffset = offset + i / ConsumeQueue.UNIT_SIZE;
            messageResult.setNextOffset(nextOffset);

            return messageResult;
        }
        return new GetMessageResult(GetMessageStatus.NOT_FOUND, 0, 0);
    }

    public ConcurrentMap<String, Set<BrokerEntity>> getClusterTables() {
        return clusterTables;
    }

    public void addClusterTables(String cluster, Set<BrokerEntity> set) {
        this.clusterTables.put(cluster, set);
    }

    public BrokerEntity getMaster() {
        Set<BrokerEntity> set = this.clusterTables.get(brokerController.getCluster_name());

        if (set != null) {
            for (BrokerEntity brokerEntity : set) {
                if (brokerEntity.getBroker_id() == 0) {
                    return brokerEntity;
                }
            }
        }
        return null;
    }

    class FlushQueueService extends ServerThread {

        long lastFlushTime = 0;

        @Override
        public String getServiceName() {
            return FlushQueueService.class.getName();
        }

        @Override
        public void run() {
            while (!isStopped()) {
                try {
                    this.waitForRunning(1000);//等待1秒
                    flush(1);
                } catch (Exception e) {
                    System.out.println(FlushQueueService.class.getName() + "刷盘异常");
                }
            }
            //全部刷盘
            flush(0);
        }

        public void flush(int retry) {
            //默认刷3页
            int flushPage = 3;
            if (retry == 0) {
                flushPage = 0;
            }
            //30秒
            int flushInteral = 30000;
            long nowTime = System.currentTimeMillis();
            //超时了
            if ((nowTime) > (lastFlushTime + flushInteral)) {
                lastFlushTime = nowTime;
                flushPage = 0;//超时的话，默认全部刷盘
            }

            for (ConcurrentMap<Integer, ConsumeQueue> consumeQueue : consumeQueueTables.values()) {
                for (ConsumeQueue queue : consumeQueue.values()) {
                    queue.flush(flushPage);
                }
            }

        }


    }

    /**
     * 逻辑队列消息存储服务
     */
    class ReputMessageService extends ServerThread {

        //在系统第一次启动的时候，要检索一次队列文件，第一次初始化完成后，reputFromOffset就是最后一条消息所在位移
        private volatile long reputFromOffset = 0l;

        public long getReputFromOffset() {
            return reputFromOffset;
        }

        public void setReputFromOffset(long reputFromOffset) {
            this.reputFromOffset = reputFromOffset;
        }

        //确保不能超过数据库文件的当前可读位置
        private boolean isCommitLogAvailable() {
            return this.reputFromOffset < DefaultMessageStore.this.commitLog.getMaxOffset();
        }

        @Override
        public String getServiceName() {
            return ReputMessageService.class.getSimpleName();
        }

        public void doReput() {
            for (boolean doNext = true; this.isCommitLogAvailable() && doNext; ) {

                /**
                 * 服务每次启动后，都会从commitlog文件挨个遍历消息，如果consumequeue逻辑文件不存在，那么恢复逻辑文件数据；
                 */
                SelectMappedBufferResult result = DefaultMessageStore.this.commitLog.getData(reputFromOffset);
                if (result != null) {
                    try {
                        //下次开始读取的最小位置
                        this.reputFromOffset = result.getStartOffset();
                        //消息大小
                        int rSize = result.getSize();
                        //遍历消息
                        for (int readSize = 0; readSize < rSize && doNext; ) {
                            //逐条解析消息，然后将关键信息存储到consumequeue中
                            DispatchRequest dispatchRequest = DefaultMessageStore.this.commitLog.checkMessageAndReturnSize(result.getByteBuffer(), false, false);
                            if (dispatchRequest != null) {
                                //消息大小
                                int size = dispatchRequest.getMsgSize();
                                if (size >= 0) {
                                    //存储队列
                                    if (!dispatchRequest.getTopic().equals("exec")) {
                                        DefaultMessageStore.this.doDispatch(dispatchRequest);
                                    }
                                    this.reputFromOffset += size;
                                    readSize += size;
                                } else {
                                    doNext = false;
                                }
                            } else {
                                doNext = false;
                            }
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                    } finally {
                        result.release();
                    }
                } else {
                    doNext = false;
                }

            }
        }

        @Override
        public void run() {
            while (!this.isStopped()) {
                try {
                    Thread.sleep(10);
                    doReput();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

    public MessageStoreConfig getMessageStoreConfig() {
        return messageStoreConfig;
    }

    public Controller getBrokerController() {
        return brokerController;
    }

    public CommitLog getCommitLog() {
        return commitLog;
    }
}
